package com.whcoding.rabbitmq.example.consumer;

import com.rabbitmq.client.Channel;
import com.whcoding.rabbitmq.example.config.RabbitConfig;
import com.whcoding.rabbitmq.example.deal.DealMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
public class ConsumerHandler {

    private static final Logger logger = LoggerFactory.getLogger(ConsumerHandler.class);

    /**
     * 配置线程池
     */
    private ExecutorService threadPool = Executors.newFixedThreadPool(64);


    /**
     * 处理消息
     */
    @Autowired
    DealMessage dealMessage;


    /**
     * 订单发送通知MQ
     *
     * @param message
     * @param deliveryTag
     * @param channel
     * @throws IOException
     */
    @RabbitHandler
    //也可监听多个队列：@RabbitListener(queues = {"hello", "hi"})
    @RabbitListener(queues = RabbitConfig.ORDER_CODE_SEND_QUEUE)
    public void process1(@Payload String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        logger.info("从MQ获取的数据为:{}", message);
        try {
            //使用多线程处理数据
            threadPool.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        //消息处理
                        dealMessage.dealOrderCodeSend(message);
                        channel.basicAck(deliveryTag, false);
                    } catch (IOException e) {
                        logger.error(e.getMessage());
                    }
                }
            });
        } catch (Exception e) {
            try {
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ex) {
                logger.error("从MQ获取的数据异常：{}", message, ex);
            }
            logger.error("从MQ获取的数据异常:{}", message, e);
        }
    }


    /**
     * 签收订单通知MQ
     *
     * @param message
     * @param deliveryTag
     * @param channel
     * @throws IOException
     */
    @RabbitHandler
    @RabbitListener(queues = RabbitConfig.ORDER_CODE_SIGN_QUEUE)
    public void process2(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        logger.info("从MQ获取的数据为:{}", message);

        //byte数组转字符串
        String messageStr = new String(message.getBody(), StandardCharsets.UTF_8);
        if (StringUtils.isEmpty(messageStr)) {
            logger.warn("byte数组转字符串。结果为空，byte[] = {}", message);
            return;
        }
        try {

            //使用多线程处理数据
            threadPool.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        //消息处理
                        dealMessage.dealOrderCodeSign(messageStr);
                        channel.basicAck(deliveryTag, false);
                    } catch (IOException e) {
                        logger.error(e.getMessage());
                    }
                }
            });
        } catch (Exception e) {
            try {
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ex) {
                logger.error("从MQ获取的Sign数据异常：{}", message, ex);
            }
            logger.error("从MQ获取的Sign数据异常:{}", message, e);
        }
    }


}
